package com.pj.project4sp.sp_mqqt_log;


import com.baomidou.mybatisplus.core.toolkit.CollectionUtils;

import com.pj.utils.LogUtil;
import com.pj.utils.sg.AjaxJson;
import com.pj.utils.sg.NbUtil;
import net.dreamlu.iot.mqtt.codec.MqttPublishMessage;
import net.dreamlu.iot.mqtt.codec.MqttQoS;
import net.dreamlu.iot.mqtt.core.server.event.IMqttMessageListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
import org.tio.core.ChannelContext;

import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.concurrent.*;

@Service
public class MqttServerMessageListener implements IMqttMessageListener {
    private static final Logger logger = LoggerFactory.getLogger(MqttServerMessageListener.class);
    private static final List<SpMqqtLog> spMqqtLogs = new ArrayList<>();
    @Autowired
    SpMqqtLogMapper spMqqtLogMapper;

    @Override
    public void onMessage(ChannelContext context, String clientId, String topic, MqttQoS qoS, MqttPublishMessage message) {
//        logger.info("clientId:{} message:{} payload:{}", clientId, message, new String(message.getPayload(), StandardCharsets.UTF_8));
//        JSONObject json = JSONObject.parseObject(new String(message.getPayload(), StandardCharsets.UTF_8));
        SpMqqtLog spMqqtLog = new SpMqqtLog();
        spMqqtLog.setId(SnowflakeIdGenerator.nextId());
        spMqqtLog.setClientId(clientId);
        spMqqtLog.setChannel(topic);
        spMqqtLog.setQoS(qoS.value());
        spMqqtLog.setMqttMessage(String.valueOf(message));
        spMqqtLog.setMsg(new String(message.getPayload(), StandardCharsets.UTF_8));
        spMqqtLog.setCreateTime(new Date());
        spMqqtLogs.add(spMqqtLog);


    }
    @Transactional(rollbackFor = Exception.class, propagation= Propagation.REQUIRED)
    public AjaxJson run() {
        if (spMqqtLogs.size() <= 0) {
//            logger.info("===== 无数据 =====");
            return AjaxJson.getError("无数据");
        }
        logger.info("===== 开始插入数据 =====");
        long startTime = System.currentTimeMillis();
        try {
            List list = new ArrayList<>();
            int maxCount = 1000;
            if (spMqqtLogs.size() < maxCount) {
                maxCount = spMqqtLogs.size();
            }
            for (int i = 0; i < maxCount; i++) {
                list.add(spMqqtLogs.get(0));
                spMqqtLogs.remove(0);
            }
            logger.info("开始：" + (System.currentTimeMillis() - startTime) + "毫秒");
            int insertBatchSql = spMqqtLogMapper.insertBatchSql(list);
            list.clear();
            logger.info("结束：" + (System.currentTimeMillis() - startTime) + "毫秒");
            logger.info("插入："+insertBatchSql);
            logger.info("总数："+spMqqtLogs.size());

        } catch (Exception e) {
            e.printStackTrace();
        }
        return AjaxJson.getError("添加失败");
    }


}